package com.afdor.rws.event.channel.aliyun.mq;

import com.afdor.rws.annotation.extension.Extension;
import com.afdor.rws.core.extension.ExtensionLoader;
import com.afdor.rws.event.Serialization;
import com.afdor.rws.event.channel.AbstractChannelProvider;
import com.afdor.rws.event.channel.ChannelProvider;
import com.afdor.rws.event.channel.client.consumer.EventConsumer;
import com.afdor.rws.event.channel.client.consumer.handler.EventHandler;
import com.afdor.rws.event.channel.client.consumer.handler.EventRetryStrategy;
import com.afdor.rws.event.channel.client.producer.EventProducer;
import com.afdor.rws.event.config.EventConfigUtils;
import lombok.EqualsAndHashCode;
import lombok.NoArgsConstructor;
import lombok.NonNull;
import lombok.ToString;

/**
 * 基于阿里云MNS的事件消息通道实现
 *
 * @author 悭梵
 * @date Created in 2018-08-08 10:38
 */
@ToString
@EqualsAndHashCode
@NoArgsConstructor
@Extension(value = "amq", order = Extension.NORMAL_ORDER - 100, interfaceClass = ChannelProvider.class)
public class AliyunMQChannelProvider extends AbstractChannelProvider {

    /**
     * 阿里云账户标识
     */
    public static final String ALIYUN_MQ_SECRET_ID = "rws.event.channel.aliyun.mns.secretId";
    /**
     * MNS API秘钥
     */
    public static final String ALIYUN_MQ_SECRET_KEY = "rws.event.channel.aliyun.mns.secretKey";
    /**
     * MNS的接入地址
     */
    public static final String ALIYUN_MQ_ENDPOINT = "rws.event.channel.aliyun.mns.endpoint";

    /**
     * 生产者
     */
    private volatile EventProducer eventProducer;

    @Override
    public AliyunMQChannelProvider init() {
        return this;
    }

    @NonNull
    @Override
    protected EventConsumer createEventConsumer(@NonNull String eventSourceName, @NonNull EventHandler eventHandler) {
        String eventConsumerName = EventConfigUtils.getProperty(EventConfigUtils.EVENT_CONSUMER_NAME, "amq_topic");
        String eventRetryName = EventConfigUtils.getProperty(EventConfigUtils.EVENT_CONSUMER_RETRY_NAME, "amq");
        String serializationName = EventConfigUtils.getProperty(EventConfigUtils.EVENT_CHANNEL_SERIALIZATION_NAME, "fastjson");
        /* 构建并设置消费者对象 */
        EventConsumer eventConsumer = ExtensionLoader
                .getInstance(EventConsumer.class)
                .newExtension(eventConsumerName)
                .setEventHandler(eventHandler)
                .setSerialization(ExtensionLoader
                        .getInstance(Serialization.class)
                        .getExtensionOrDefault(serializationName))
                .setEventSourceName(eventSourceName);
        /* 重试策略 */
        ExtensionLoader<EventRetryStrategy> eventRetryStrategyExtensionLoader = ExtensionLoader.getInstance(EventRetryStrategy.class);
        if (eventRetryStrategyExtensionLoader.hasExtension(eventRetryName)) {
            if (eventRetryStrategyExtensionLoader.hasConstructor(eventRetryName, eventConsumer)) {
                eventConsumer.setEventRetryStrategy(eventRetryStrategyExtensionLoader.newExtension(eventRetryName, eventConsumer));
            } else {
                eventConsumer.setEventRetryStrategy(eventRetryStrategyExtensionLoader.newExtension(eventRetryName));
            }
        }
        return eventConsumer;
    }

    @NonNull
    @Override
    protected EventProducer getEventProducer() {
        if (eventProducer == null) {
            String eventProduceName = EventConfigUtils.getProperty(EventConfigUtils.EVENT_PRODUCER_NAME, "amq_topic");
            String serializationName = EventConfigUtils.getProperty(EventConfigUtils.EVENT_CHANNEL_SERIALIZATION_NAME, "fastjson");
            eventProducer = ExtensionLoader
                    .getInstance(EventProducer.class)
                    .newExtension(eventProduceName)
                    .setSerialization(ExtensionLoader
                            .getInstance(Serialization.class)
                            .getExtensionOrDefault(serializationName))
                    .init();
        }
        return eventProducer;
    }
}
